package com.nageoffer.shortlink.project.config;

import lombok.AllArgsConstructor;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.ThreadPoolExecutor;

@Configuration
@RequiredArgsConstructor
public class KafkaListenerConfig {

    private final ConsumerFactory<String, Object> consumerFactory;



    @Bean(name = "kafkaHandlerExecutor")
    public ThreadPoolTaskExecutor kafkaHandlerExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

        // 核心配置参数
        executor.setCorePoolSize(2);       // 常驻线程数
        executor.setMaxPoolSize(2);         // 最大线程数
        executor.setThreadNamePrefix("kafka-handler-"); // 线程名前缀

        // 拒绝策略: 当队列满时由调用线程执行任务
        executor.setRejectedExecutionHandler(
                new ThreadPoolExecutor.DiscardOldestPolicy()
        );
        executor.initialize();
        return executor;
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(@Qualifier("kafkaHandlerExecutor")ThreadPoolTaskExecutor kafkaHandlerExecutor) {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
                new ConcurrentKafkaListenerContainerFactory<>();

        factory.setConsumerFactory(consumerFactory);

        // 设置消费线程数（通常等于分区数）
        factory.setConcurrency(2);

        // 设置处理线程池
        factory.getContainerProperties().setListenerTaskExecutor(kafkaHandlerExecutor);

        // 可选：批量消费配置
        factory.setBatchListener(true);

        // 可选：设置ACK模式
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);

        return factory;
    }



}
